package com.cyj.kafka.broadcast;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @ClassName BroadcastProducer
 * @Description 广播生产者
 * @Author chixiaowai
 * @Date 2022/10/23 22:25
 * @Version 1.0
 */

public class BroadcastProducer {
    public static void main(String[] args) {
        //配置
        Properties properties = new Properties();
        //连接集群
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop101:9092,hadoop102:9092,hadoop103:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        //创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        try {
            //发送数据
            for (int i = 1; i <= 50; i++) {
                kafkaProducer.send(new ProducerRecord<String, String>("second", "生产第" + i + "条消息到second"), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if (e == null){
                            System.out.println("主题:" + recordMetadata.topic() + "\t" + "分区:" + recordMetadata.partition());
                        }
                    }
                });
                kafkaProducer.send(new ProducerRecord<String, String>("first", "生产第" + i + "条消息到first"), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if (e == null){
                            System.out.println("主题:" + recordMetadata.topic() + "\t" + "分区:" + recordMetadata.partition());
                        }
                    }
                });
                Thread.sleep(2);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            //关闭资源
            kafkaProducer.close();
        }
    }
}
